#include "pch.h"
#include "Threadpool.h"
#include "..\Common\SRWLock.h"

CThreadpool::CThreadpool(UINT nMinThread, UINT nMaxThread)
	: m_nMinThread(nMinThread)
	, m_nMaxThread(nMaxThread)
	, m_nThreadNum(0)
	, m_pTpWork(NULL)
{
	AssertInUIThread();
}

CThreadpool::~CThreadpool()
{
	AssertInUIThread();
	assert(m_pTpWork == NULL);
	assert(m_deqTasks.empty());
}

bool CThreadpool::Init()
{
	AssertInUIThread();

	m_pTpWork = CreateThreadpoolWork([](PTP_CALLBACK_INSTANCE, PVOID pContext, PTP_WORK)
	{
		((CThreadpool *)pContext)->RunTask();
	}, this, NULL);

	return m_pTpWork != NULL;
}

void CThreadpool::Exit()
{
	AssertInUIThread();

	if (m_pTpWork == NULL)
		return;

	// Notify all threads to exit
	CExclusiveLock lock(m_SRWLock);
	auto deqTasks = std::move(m_deqTasks);
	lock.Unlock();

	// Wait for all threads to exit
	WaitForThreadpoolWorkCallbacks(m_pTpWork, TRUE);
	CloseThreadpoolWork(m_pTpWork);
	m_pTpWork = NULL;

	// Sequentially clean up pending tasks
	while (deqTasks.size())
	{
		deqTasks.pop_front();
	}
}

bool CThreadpool::PostTask(ICallbackPtr&& pTask)
{
	CExclusiveLock lock(m_SRWLock);
	m_deqTasks.emplace_back(std::move(pTask));
	bool bNewThread = false;

	// Less than the number of core threads, or the number of tasks is more than half of the number of threads, start a new thread
	if (m_nThreadNum < m_nMinThread || m_nThreadNum < m_nMaxThread && m_nThreadNum < m_deqTasks.size() * 2)
	{
		m_nThreadNum++;
		bNewThread = true;
	}

	lock.Unlock();

	if (bNewThread)
		SubmitThreadpoolWork(m_pTpWork);

	return true;
}

void CThreadpool::RunTask()
{
	AssertInThreadpool();

	CExclusiveLock lock(m_SRWLock);

	while (m_deqTasks.size())
	{
		auto pTask = std::move(m_deqTasks.front());
		m_deqTasks.pop_front();
		lock.Unlock();

		pTask->Invoke();
		pTask = nullptr;

		lock.Lock();
	}

	m_nThreadNum--;
}
